package com.deep.flink.mysql;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class MySQLBinlog {

    public static void main(String[] args) {
        //source
        SourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("127.0.0.1")
                .port(3306)
                .username("root")
                .password("root")
                .serverTimeZone("UTC")
                .databaseList("ds_canal")
                .tableList("user_base")
                .deserializer(new StringDebeziumDeserializationSchema())
                .build();
        //exe env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> streamSource = env.addSource(sourceFunction);
        streamSource.print();
//        streamSource.addSink(new MySQLSinkFunction());
        //执行
        try {
            env.execute("test");
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
}
